package com.kafka.learn.kafka.first;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;


import java.util.Collections;

import java.util.Properties;

/**
 * @className ConsumerTask
 * @description Kafka消费者------死循环监听
 * @author 郭辉
 * @date 2021/12/23 20:28
 * @company 海康威视
 * @since 1.0
 */
@Component()
@Scope("prototype")
public class ConsumerTask extends Thread {

    /**
     * kafka的IP端口
     * */
    public static final String BROKER_LIST = "localhost:9092";

    /**
     * 订阅的topic
     * */
    public static final String TOPIC = "SNAP_IMAGE_INFO_TOPIC";

    /**
     * 消费者组
     * */
    public static final String GROUP_ID = "group.demo";

    /**
     * 线程运行方法
     * */
    @Override
    public void run() {
        final Properties properties = new Properties();
        /**
         * 键反序列化
         * */
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        /**
         * 值得反序列化
         * */
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        /**
         *集群地址
         * */
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);
        /**
         * 消费组
         * */
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);

        /**
         * 实例化消费者（用于消息的接受）
         * */
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);

        /**
         * 订阅topic
         * */
        consumer.subscribe(Collections.singleton(TOPIC));

        /**
         * 轮询监听kafka生产者
         * */
        while (true) {
            // 在100ms内等待Kafka的broker返回数据.超时参数指定poll在多久之后可以返回，不管有没有可用的数据都要返回
            final ConsumerRecords<String, String> records = consumer.poll(1000);

            for (ConsumerRecord<String, String> record : records) {
                // 解析消息
                final String value = record.value();
                System.out.println(value);
            }
        }
    }

    public static void main(String[] args) {
        new ConsumerTask().start();
    }
}
